Add large-payload blob auto-purge (opt-in singleton job, worker/SDK side) by YunchuWang · Pull Request #758 · microsoft/durabletask-dotnet

YunchuWang · 2026-07-08T19:51:11Z

Summary

Large orchestration payloads are externalized to Azure Blob Storage by the AzureBlobPayloads extension as blob:v1:<container>:<blobName> tokens. The DTS backend stores those tokens but cannot delete the backing blobs — it has no storage credentials; only this SDK can. This PR implements the worker/SDK side of large-payload blob auto-purge.

Design (opt-in singleton durable entity + orchestration job)

Instead of an always-on background stream, this mirrors the existing src/ExportHistory feature: an opt-in, whole-scheduler singleton durable entity + orchestration job that drains soft-deleted payload rows the backend exposes and deletes their blobs, then acks so the backend can hard-delete the rows.

PayloadStore.DeleteAsync — added as a virtual method (default throws NotSupportedException, so it is non-breaking for existing external subclasses). BlobPayloadStore overrides it to decode the token and call DeleteIfExistsAsync (idempotent — deleting a missing blob is a no-op).
BlobPurgeJob (TaskEntity singleton) — Create is a no-op when already Active so racing client processes don't disturb the running job (intentionally softer than ExportJob.Create, which throws). Run starts a fixed-instance-id orchestrator.
BlobPurgeJobOrchestrator (perpetual) — each cycle: fetch a batch of tombstones, delete the blobs with capped parallelism (32), ack only the successful deletions (failed tokens stay tombstoned to retry), idle on a 1-minute timer when there's nothing to purge, and ContinueAsNew every 5 cycles to keep history small. Activities use a small retry policy.
ExecuteBlobPurgeJobOperationOrchestrator — client → entity bridge (mirrors export).
Activities (constructor DI): GetTombstonedPayloadsActivity, DeleteExternalBlobActivity (returns false + logs on failure so one bad token can't fail the batch), AckPurgedPayloadsActivity.
Purge RPCs on the core client — the worker fetch/ack RPCs are now first-class methods on DurableTaskClient (GetTombstonedPayloadsAsync/AckPurgedPayloadsAsync), overridden in GrpcDurableTaskClient — mirroring how ExportHistory added ListInstanceIdsAsync/GetOrchestrationHistoryAsync. The purge activities inject DurableTaskClient directly; no dedicated gRPC client is needed. AzureManaged reuses GrpcDurableTaskClient, so it inherits these methods with no extra client.
Options — LargePayloadStorageOptions.AutoPurge (opt-in, default false) and PayloadPurgeBatchSize (default 500).
Enablement — client-side BlobPurgeJobStarter (IHostedService) ensures the singleton job when AutoPurge is enabled, on a background task that does not block host startup and retries until the backend is reachable. The worker always registers the entity/orchestrators/activities (not gated on AutoPurge) so a client-enabled job always has something to execute.

gRPC contract

Two new unary RPCs added to TaskHubSidecarService in src/Grpc/orchestrator_service.proto (worker is the client; wire paths /TaskHubSidecarService/GetTombstonedPayloads and /AckPurgedPayloads):

rpc GetTombstonedPayloads(GetTombstonedPayloadsRequest) returns (GetTombstonedPayloadsResponse);
rpc AckPurgedPayloads(AckPurgedPayloadsRequest) returns (AckPurgedPayloadsResponse);

message TombstonedPayload { int32 partitionId = 1; int64 instanceKey = 2; int64 payloadId = 3; string token = 4; }
message PayloadPurgeAck   { int32 partitionId = 1; int64 instanceKey = 2; int64 payloadId = 3; }
message GetTombstonedPayloadsRequest  { int32 limit = 1; }
message GetTombstonedPayloadsResponse { repeated TombstonedPayload payloads = 1; }
message AckPurgedPayloadsRequest      { repeated PayloadPurgeAck acks = 1; }
message AckPurgedPayloadsResponse     { }

C# stubs are generated at build time by Grpc.Tools (not committed). The authoritative proto change is a follow-up in microsoft/durabletask-protobuf#76; once it merges, src/Grpc/orchestrator_service.proto should be re-synced from upstream (content identical to what's here). The vendored change lets this PR build/test standalone.

Testing

dotnet build Microsoft.DurableTask.sln — succeeds, 0 errors.
dotnet test test/AzureBlobPayloads.Tests — 11 passed (BlobPayloadStore delete/idempotency + BlobPurgeJob.Create no-op-when-Active + options defaults).
dotnet test test/ExportHistory.Tests — 147 passed (shared patterns unaffected).
dotnet test test/Client/Core.Tests + test/Client/Grpc.Tests — 43 + 47 passed (core DurableTaskClient / GrpcDurableTaskClient edits).

Notes / deviations

BlobPurgeJobStatus.Pending is the 0/default value (mirroring ExportJobStatus.Pending=0) so a freshly initialized entity never appears Active.
Added a RecordPurged entity op to track a cumulative PurgedCount.

Draft — not for merge until the authoritative proto PR lands and the backend serves the RPCs on TaskHubSidecarService.

Co-authored-by: Copilot App 223556219+Copilot@users.noreply.github.com

Large orchestration payloads are externalized to Azure Blob Storage as `blob:v1:<container>:<blobName>` tokens. The DTS backend stores those tokens but cannot delete the backing blobs (it has no storage credentials) — only this SDK can. This adds an opt-in, whole-scheduler singleton durable entity + orchestration job (mirroring src/ExportHistory) that drains payload rows the backend has soft-deleted and deletes their blobs, then acks so the backend can hard-delete the rows. Design: - PayloadStore.DeleteAsync is virtual (default throws NotSupportedException so it is non-breaking for existing external subclasses); BlobPayloadStore overrides it to decode the token and call DeleteIfExistsAsync (idempotent). - BlobPurgeJob (TaskEntity singleton): Create is a no-op when already Active so racing client processes don't disturb the running job; Run starts a fixed-id orchestrator. - BlobPurgeJobOrchestrator (perpetual): fetch a batch of tombstones, delete the blobs with capped parallelism, ack the successful deletions (failed tokens stay tombstoned to retry), idle on a timer when empty, ContinueAsNew periodically. - ExecuteBlobPurgeJobOperationOrchestrator bridges client -> entity. - Two new unary RPCs on TaskHubSidecarService: GetTombstonedPayloads / AckPurgedPayloads (authoritative proto follow-up: microsoft/durabletask-protobuf#76). - LargePayloadStorageOptions gains AutoPurge (opt-in, default false) and PayloadPurgeBatchSize (default 500). - Client-side BlobPurgeJobStarter (IHostedService) ensures the singleton job when AutoPurge is enabled, without blocking host startup. Worker always registers the entity/orchestrators/activities so a client-enabled job has something to run. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

YunchuWang · 2026-07-13T22:50:15Z

+/// <param name="PartitionId">The backend partition that owns the payload row.</param>
+/// <param name="InstanceKey">The orchestration instance key the payload belongs to.</param>
+/// <param name="PayloadId">The backend identifier of the soft-deleted payload row.</param>
+public sealed record PayloadPurgeAckDto(int PartitionId, long InstanceKey, long PayloadId);


remove suffix dto?

YunchuWang · 2026-07-13T22:50:19Z

+/// <param name="InstanceKey">The orchestration instance key the payload belongs to.</param>
+/// <param name="PayloadId">The backend identifier of the soft-deleted payload row.</param>
+/// <param name="Token">The externalized payload token whose backing blob should be deleted.</param>
+public sealed record TombstonedPayloadDto(int PartitionId, long InstanceKey, long PayloadId, string Token);


remove suffix dto?

YunchuWang · 2026-07-13T22:51:40Z

+    public override async Task<List<TombstonedPayloadDto>> GetTombstonedPayloadsAsync(
+        int limit, CancellationToken cancellation = default)
+    {
+        P.GetTombstonedPayloadsResponse response = await this.sidecarClient.GetTombstonedPayloadsAsync(


lets add limit range check, not here whenever limit is specified, > 0 and < 1000

YunchuWang · 2026-07-13T22:56:05Z

+        LargePayloadStorageOptions opts = this.options.Get(this.builderName);
+        if (!opts.AutoPurge)
+        {
+            this.logger.BlobPurgeDisabled();
+            return Task.CompletedTask;


is this a defensive check? since this starter class should be conditionally di based on auto purge flag before checking here right?

YunchuWang · 2026-07-13T23:05:18Z

+                    this.entityId, cancellation: cancellationToken);
+            return metadata is not null && metadata.State.Status == BlobPurgeJobStatus.Active;
+        }
+        catch (NotSupportedException)


double check please, it should be supported!

YunchuWang · 2026-07-13T23:06:06Z

+    /// The job is not running. This is the default status of a freshly initialized entity, so it is kept
+    /// as the zero value to avoid a brand-new entity accidentally appearing active.
+    /// </summary>
+    Stopped,


change to pending

YunchuWang · 2026-07-13T23:06:32Z

+    /// <summary>
+    /// The job has failed.
+    /// </summary>
+    Failed,


when this will fail, this should always run

YunchuWang · 2026-07-13T23:10:47Z

+        {
+            try
+            {
+                if (await this.IsJobActiveAsync(cancellationToken))


unncessary right and cant guard against race condition?
we can ensure one entity instance for auto purge by using one unique entityt job id right?

YunchuWang · 2026-07-13T23:31:16Z

+                    nameof(BlobPurgeJob.Create),
+                    new BlobPurgeJobCreationOptions(batchSize));
+
+                await this.client.ScheduleNewOrchestrationInstanceAsync(


just scheduled is enough?

YunchuWang · 2026-07-13T23:32:31Z

+            {
+                return;
+            }
+            catch (Exception ex) when (ex is not OutOfMemoryException and not StackOverflowException)


did you check if its this job already exist case whether it throws exception, if not we do not want to retry

YunchuWang · 2026-07-13T23:34:59Z

+
+    async Task EnsureJobAsync(int batchSize, CancellationToken cancellationToken)
+    {
+        while (!cancellationToken.IsCancellationRequested)


why this is looping forever, what i want is start the job once if not exist, and there should only be one job running at any time, and if the job failed only you should retry scheduling it

YunchuWang · 2026-07-13T23:41:51Z

+/// <param name="PurgeBatchSize">
+/// The maximum number of tombstoned payloads to request from the backend per cycle.
+/// </param>
+public sealed record BlobPurgeJobCreationOptions(int PurgeBatchSize);


really need this record just one param?

YunchuWang · 2026-07-13T23:51:29Z

+    /// <inheritdoc/>
+    public override async Task<List<TombstonedPayloadDto>> RunAsync(TaskActivityContext context, int input)
+    {
+        int limit = input > 0 ? input : 500;


magic number

YunchuWang · 2026-07-13T23:53:29Z

+/// <param name="store">The payload store used to delete blobs.</param>
+/// <param name="logger">The logger instance.</param>
+[DurableTask]
+public class DeleteExternalBlobActivity(


should we delete a batch instead of one at at time per activity call

YunchuWang · 2026-07-14T00:17:05Z

+    public static IDurableTaskClientBuilder UseExternalizedPayloads(
+        this IDurableTaskClientBuilder builder,
+        Action<LargePayloadStorageOptions> configure)
+    {
+        Check.NotNull(builder);
+        Check.NotNull(configure);
+
+        builder.Services.Configure(builder.Name, configure);
+        builder.Services.AddSingleton<PayloadStore>(sp =>
+        {
+            LargePayloadStorageOptions opts = sp.GetRequiredService<IOptionsMonitor<LargePayloadStorageOptions>>().Get(builder.Name);
+            return new BlobPayloadStore(opts);
+        });
+
+        return UseExternalizedPayloadsCore(builder);


why we need configure blobpayloadstore in client

…er simplification - Drop the `Dto` suffix now that the payload records are first-class public types in `Microsoft.DurableTask.Client` (`TombstonedPayload`, `PayloadPurgeAck`). - Collapse the magic `500` batch-size literal into a single `BlobPurgeConstants.DefaultBatchSize` used everywhere. - Rename `BlobPurgeJobStatus.Stopped` -> `Pending` (still the zero value) and remove the dead `Failed` member (nothing ever set it; the job self-heals). - Make the perpetual orchestrator self-heal: wrap each cycle in try/catch so a transient backend/entity/activity failure logs, backs off, and continues instead of failing the orchestration and killing the eternal loop. - Ack poison tokens: `DeleteExternalBlobActivity` now returns a three-way `BlobDeleteResult` (Deleted/Discarded/Retry). Malformed tokens are discarded and acked so the backend can clear the stuck row instead of re-streaming it forever; transient failures stay tombstoned to retry. - Replace the single-value `BlobPurgeJobCreationOptions` record with a plain `int` on `BlobPurgeJob.Create`. - Guard the client fetch RPC: `GetTombstonedPayloadsAsync` throws `ArgumentOutOfRangeException` unless `0 < limit < 1000`. - Simplify `BlobPurgeJobStarter` to a fixed-instance-id fire-once: drop the entity-active pre-check and schedule the Create bridge once with a fixed instance id, retrying only until the backend is reachable. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

+        foreach (DeleteOutcome outcome in outcomes)
+        {
+            // Acknowledge blobs that were deleted (or already gone) and poison tokens that can never succeed
+            // so the backend can hard-delete their rows; transient failures stay tombstoned to retry.
+            if (outcome.ShouldAck)
+            {
+                acks.Add(outcome.Ack);
+            }
+        }


YunchuWang force-pushed the yunchuwang-wangbill-blob-payload-autopurge-sdk branch 2 times, most recently from 82ae04d to 0ac2dc3 Compare July 8, 2026 20:50

github-code-quality Bot found potential problems Jul 8, 2026

View reviewed changes

Comment thread src/Extensions/AzureBlobPayloads/AutoPurge/BlobPayloadAutoPurgeService.cs Fixed

YunchuWang force-pushed the yunchuwang-wangbill-blob-payload-autopurge-sdk branch 4 times, most recently from 0752610 to cab0e9a Compare July 13, 2026 19:07

YunchuWang changed the title ~~Add large-payload blob auto-purge (worker/SDK side)~~ Add large-payload blob auto-purge (opt-in singleton job, worker/SDK side) Jul 13, 2026

github-code-quality Bot found potential problems Jul 13, 2026

View reviewed changes

Comment thread src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs Fixed

YunchuWang force-pushed the yunchuwang-wangbill-blob-payload-autopurge-sdk branch from cab0e9a to c05b15a Compare July 13, 2026 20:38

YunchuWang force-pushed the yunchuwang-wangbill-blob-payload-autopurge-sdk branch from c05b15a to 306d19f Compare July 13, 2026 22:43

YunchuWang commented Jul 13, 2026

View reviewed changes

YunchuWang commented Jul 14, 2026

View reviewed changes

github-code-quality Bot found potential problems Jul 14, 2026

View reviewed changes

Uh oh!

Conversation

YunchuWang commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Design (opt-in singleton durable entity + orchestration job)

gRPC contract

Testing

Notes / deviations

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

YunchuWang Jul 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

YunchuWang Jul 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

YunchuWang commented Jul 8, 2026 •

edited

Loading

YunchuWang Jul 13, 2026 •

edited

Loading

YunchuWang Jul 13, 2026 •

edited

Loading